package com.yuan;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.Scanner;

public class ExampleForHBase {

    // HBase配置对象
    public static Configuration configuration;
    // HBase连接对象
    public static Connection connection;
    // HBase管理对象，用于执行DDL操作（如创建表）
    public static Admin admin;

    public static void main(String[] args) throws IOException {
        String tableName = "Student";
        String rowKey = "Tom";
        while(true){
            Scanner sc = new Scanner(System.in);
            System.out.println("-------------------------------");
            System.out.println("请输入要执行的操作：");
            System.out.println("1. 创建表");
            System.out.println("2. 添加记录");
            System.out.println("3. 扫描 Score:Math 列");
            System.out.println("4. 修改 Math 成绩");
            System.out.println("5. 删除行");
            System.out.println("0. 退出");
            int op = sc.nextInt();
            switch (op){
                case 1:
                    // 1. 创建表
                    String[] columnFamilies = {"Info", "Score"};
                    System.out.println("正在创建表...");
                    createTable(tableName, columnFamilies);
                    break;
                case 2:
                    // 2. 添加记录
                    String[] fields = {"Info:Name", "Score:Math", "Score:Computer Science", "Score:English"};
                    String[] values = {"Tom", "90", "85", "95"};
                    System.out.println("正在添加记录...");
                    addRecord(tableName, rowKey, fields, values);
                    break;
                case 3:
                    // 3. 扫描 Score:Math 列
                    System.out.println("正在扫描 Score:Math 列...");
                    scanColumn(tableName, "Score:Math");
                    break;
                case 4:
                    // 4. 修改 Math 成绩
                    System.out.println("正在修改记录...");
                    System.out.println("请输入新的 Math 成绩：");
                    float score = sc.nextFloat();
                    modifyData(tableName, rowKey, "Score:Math", String.valueOf(score));
                    System.out.println("修改后重新扫描 Score:Math 列...");
                    scanColumn(tableName, "Score:Math");
                    break;
                case 5:
                    // 5. 删除行
                    System.out.println("正在删除行 'Tom'...");
                    deleteRow(tableName, rowKey);
                    // 6. 再次扫描确认删除
                    System.out.println("删除后再次扫描 Score 列族数据...");
                    scanColumn(tableName, "Score");
                    break;
                default:
                    return;
            }
        }
    }

    /**
     * 初始化HBase配置和连接
     */
    public static void init() {
        // 创建HBase配置对象
        configuration = HBaseConfiguration.create();
        // 设置HBase的根目录路径
        configuration.set("hbase.rootdir", "hdfs://localhost:9000/hbase");

        try {
            // 创建HBase连接
            connection = ConnectionFactory.createConnection(configuration);
            // 获取Admin对象，用于管理表
            admin = connection.getAdmin();
        } catch (IOException e) {
            // 捕获并打印异常信息
            e.printStackTrace();
        }
    }

    /**
     * 关闭HBase连接和资源
     */
    public static void close() {
        try {
            // 关闭Admin对象
            if (admin != null) {
                admin.close();
            }
            // 关闭连接
            if (connection != null) {
                connection.close();
            }
        } catch (IOException e) {
            // 捕获并打印异常信息
            e.printStackTrace();
        }
    }

    /**
     * 创建HBase表
     *
     * @param myTableName 表名
     * @param colFamily   列族数组
     * @throws IOException 如果表创建失败或发生其他IO错误
     */
    public static void createTable(String myTableName, String[] colFamily) throws IOException {
        init();
        // 将表名转换为TableName对象
        TableName tableName = TableName.valueOf(myTableName);

        // 检查表是否已存在
        if (admin.tableExists(tableName)) {
            System.out.println("表已存在！");
        } else {
            // 构建表描述符
            TableDescriptorBuilder tableDescriptor = TableDescriptorBuilder.newBuilder(tableName);
            for (String str : colFamily) {
                // 构建列族描述符，并添加到表描述符中
                ColumnFamilyDescriptor family =
                        ColumnFamilyDescriptorBuilder.newBuilder(Bytes.toBytes(str)).build();
                tableDescriptor.setColumnFamily(family);
            }
            // 创建表
            admin.createTable(tableDescriptor.build());
            System.out.println("创建表 " + myTableName + " 成功！");
            close();
        }
    }

    public static void addRecord(String tableName,String row,String[] fields,String[] values) throws IOException {
        init();
        Table table = connection.getTable(TableName.valueOf(tableName));
        for(int i = 0;i != fields.length;i++){
            Put put = new Put(row.getBytes());
            String[] cols = fields[i].split(":");
            put.addColumn(cols[0].getBytes(), cols[1].getBytes(), values[i].getBytes());
            table.put(put);
        }
        table.close();
        close();
    }

    /**
     * 扫描HBase表中的指定列，并显示结果
     * 
     * @param tableName 表名
     * @param column 列名，可以是列族或列族:列的格式
     * @throws IOException 当与HBase的交互过程中发生通信错误时抛出
     */
    public static void scanColumn(String tableName, String column) throws IOException {
        // 初始化HBase连接
        init();
        
        // 获取指定表的实例
        Table table = connection.getTable(TableName.valueOf(tableName));
        
        // 创建Scan实例，用于配置扫描参数
        Scan scan = new Scan();
        
        // 分解列名，以处理不同的列格式
        String[] parts = column.split(":");
        if (parts.length == 1) {
            // 当只指定列族时，扫描整个列族
            scan.addFamily(Bytes.toBytes(column));
        } else if (parts.length == 2) {
            // 当指定列族和列名时，扫描特定列
            scan.addColumn(Bytes.toBytes(parts[0]), Bytes.toBytes(parts[1]));
        } else {
            // 当列格式不正确时，输出错误信息并返回
            System.out.println("列格式不正确: " + column);
            return;
        }
        
        // 执行扫描并获取扫描结果
        ResultScanner scanner = table.getScanner(scan);
        // 遍历扫描结果，显示每个单元格的信息
        for (Result result = scanner.next(); result != null; result = scanner.next()) {
            showCell(result);
        }
        
        // 关闭表实例
        table.close();
        // 关闭HBase连接
        close();
    }

        // 格式化输出Result对象中的每个单元格信息
    public static void showCell(Result result){
        Cell[] cells = result.rawCells();
        for(Cell cell:cells){
            System.out.println("行键：" + new String(Bytes.toString(cell.getRowArray(),cell.getRowOffset(), cell.getRowLength())) + " ");
            System.out.println("时间戳：" + cell.getTimestamp() + " ");
            System.out.println("列族：" + new String(Bytes.toString(cell.getFamilyArray(),cell.getFamilyOffset(), cell.getFamilyLength())) + " ");
            System.out.println("列名：" + new String(Bytes.toString(cell.getQualifierArray(),cell.getQualifierOffset(), cell.getQualifierLength())) + " ");
            System.out.println("值：" + new String(Bytes.toString(cell.getValueArray(),cell.getValueOffset(), cell.getValueLength())) + " ");
        }
    }
    
    // 修改指定表中的数据
    public static void modifyData(String tableName, String row, String column, String val) throws IOException {
        init();
        Table table = connection.getTable(TableName.valueOf(tableName));
        Put put = new Put(row.getBytes());//!构建Put对象，指定行键
    
        String[] parts = column.split(":");
        if (parts.length == 2) {
            // !为指定列族和列名添加数据
            put.addColumn(Bytes.toBytes(parts[0]), Bytes.toBytes(parts[1]), val.getBytes());
        } else {
            // 提示列格式不正确
            System.out.println("列格式不正确: " + column);
        }
    
        table.put(put);
        table.close();
        close();
    }
    
    // 删除指定表中的行
    public static void deleteRow(String tableName,String row)throws IOException{
        init();
        Table table = connection.getTable(TableName.valueOf(tableName));
        Delete delete = new Delete(row.getBytes());//!构建Delete对象，指定行键
        table.delete(delete);//!执行删除
        table.close();
        close();
    }
}